#include "stdafx.h"

#include "Socket.h"
#include "PC2DataStore.h"
#include "ParticipantExecutor.h"

#include "PC2Participant.h"


namespace PC2
{

PC2Participant::PC2Participant( NodeInfo const& aID, char const* apDBPath, NodeInfo const& aCoordinatorID, std::vector<NodeInfo> const& aParticipantIDSet)
	: PC2Node(aID, aParticipantIDSet, apDBPath)
	, m_CoordinatorInfo(aCoordinatorID)
	, m_pExecutor(NULL)
{

}

PC2Participant::~PC2Participant()
{

}

void PC2Participant::Run()
{
	while (true)
	{
		LoopForCurrentTransaction();
	}
}

void PC2Participant::LoopForCurrentTransaction()
{
	LOG("waiting for request from other node\n");

	uint32_t lID = m_pStore->GetCurrentTransId();
	TransactionInfoSet lInfoSet;
	if(0 != lID)
	{
		m_pStore->LoadTranInfoSet(lID, &lInfoSet, false);
	}
	else
	{
		lInfoSet.m_LastTrans.m_TransId = 0;
	}

	Trans lReceivedTrans;
	char lIP[32];
	unsigned short lPort;
	std::string lData;
	
	bool lSentDecesionRequest = false;
	uint32_t lCurrentDecesionRequestTransId;

	while(true)
	{
		bool lReceived = WaitForData(&lReceivedTrans, &lData, _TIMEOUT_MS * 2, lIP, &lPort);
		
		if(false == lReceived)
		{
			if(0 == lInfoSet.m_LastTrans.m_TransId
				|| EnumPrepareRejectResponse == lInfoSet.m_LastTrans.m_Type
				|| EnumCommitAck == lInfoSet.m_LastTrans.m_Type
				|| EnumAbortAck == lInfoSet.m_LastTrans.m_Type)
			{
				continue;
			}
			else if(EnumPrepareOKResponse == lInfoSet.m_LastTrans.m_Type)
			{
				//Trans lDecesionRequest;
				//asking for decision from other node
				//first coordinator
				if(false == lSentDecesionRequest)
				{
					LOG("prepare OK, waiting for action cmd timeout, request for decision\n");
					lSentDecesionRequest = true;
					SendDecisionRequest(lInfoSet.m_LastTrans.m_TransId, NULL);
					lCurrentDecesionRequestTransId = lInfoSet.m_LastTrans.m_TransId;
				}
				else //all
				{
					SendDecisionRequest(lInfoSet.m_LastTrans.m_TransId, &lInfoSet.m_RelatedNodes);
				}
			}
			continue;
		}

		if(EnumPrepareRequest == lReceivedTrans.m_Type)
		{
			LOG("received prepare request\n");
			HandlePrepareRequest(&lInfoSet, &lReceivedTrans, &lData);
		}
		else if(EnumCommitRequest == lReceivedTrans.m_Type || EnumAbortRequest == lReceivedTrans.m_Type)
		{
			LOG("received action request\n");
			HandleActionRequest(&lInfoSet, &lReceivedTrans);
			if(true == lSentDecesionRequest && lReceivedTrans.m_TransId == lCurrentDecesionRequestTransId)
			{
				lSentDecesionRequest = false;
			}
		}
		else if(EnumDecisionRequest == lReceivedTrans.m_Type)
		{
			LOG("received decision request\n");
			HandleDecisionRequest(&lInfoSet, &lReceivedTrans, lIP, lPort);
		}
	}
}

void PC2Participant::HandlePrepareRequest( TransactionInfoSet* apInfoSet, Trans* apReceivedTrans, std::string* apData )
{
	//new one
	if( 0 == apInfoSet->m_LastTrans.m_TransId || 
			( (EnumPrepareRejectResponse == apInfoSet->m_LastTrans.m_Type	
				|| EnumCommitAck == apInfoSet->m_LastTrans.m_Type
				|| EnumAbortAck == apInfoSet->m_LastTrans.m_Type) 
				&& apInfoSet->m_LastTrans.m_TransId != apReceivedTrans->m_TransId)  )
	{
		apInfoSet->m_LastTrans.m_TransId = apReceivedTrans->m_TransId;
		apInfoSet->m_LastTrans.m_NodeId = m_Info.m_Id;
		apInfoSet->m_LastTrans.m_Size = sizeof(Trans);

		NodesHeader* lpNodeHeader = (NodesHeader*)apData->c_str();
		lpNodeHeader->ReadParts(&apInfoSet->m_RelatedNodes);

		apInfoSet->m_TransData = std::string(apData->c_str() + lpNodeHeader->m_Size, apData->length() - lpNodeHeader->m_Size);

		leveldb::WriteBatch lBatch;
		bool lOK = m_pExecutor->Prepare(apInfoSet->m_LastTrans.m_TransId, apInfoSet->m_TransData.c_str(), apInfoSet->m_TransData.length(), &lBatch);
		if(lOK)
		{
			apInfoSet->m_LastTrans.m_Type = EnumPrepareOKResponse;
		}
		else
		{
			apInfoSet->m_LastTrans.m_Type = EnumPrepareRejectResponse;
		}

		PC2DataStore::WriteCurrentTransId(&lBatch, apReceivedTrans->m_TransId);
		PC2DataStore::WriteLastTransState(&lBatch, apInfoSet);
		
		if(EnumPrepareRejectResponse == apInfoSet->m_LastTrans.m_Type)
		{
			apInfoSet->m_TransData.clear();
			PC2DataStore::FinishAndClearTransactionInfo(&lBatch, apInfoSet);
		}
		else
		{
			PC2DataStore::WriteTransData(&lBatch, apInfoSet);
			PC2DataStore::WriteRelatedNodeSet(&lBatch, apInfoSet);
		}

		m_pStore->CommitBatch(&lBatch);
		
	}

	if (apInfoSet->m_LastTrans.m_TransId != apReceivedTrans->m_TransId)
	{
		LOG("received prepare req but trans_id different from current %u %u, igore\n", apReceivedTrans->m_TransId, apInfoSet->m_LastTrans.m_TransId );
		return;
	}

	if( EnumPrepareOKResponse == apInfoSet->m_LastTrans.m_Type || EnumPrepareRejectResponse == apInfoSet->m_LastTrans.m_Type)
	{
		SendTrans(&m_CoordinatorInfo, &apInfoSet->m_LastTrans);
	}
	else
	{
		LOG("received prepare req but already in action stage trans_id %u, igore\n", apInfoSet->m_LastTrans.m_TransId );
	}
}

void PC2Participant::HandleActionRequest( TransactionInfoSet* apInfoSet, Trans* apReceivedTrans )
{
	// different trans
	if(0 == apInfoSet->m_LastTrans.m_TransId || apInfoSet->m_LastTrans.m_TransId != apReceivedTrans->m_TransId)
	{
		LOG("received action req but not in action stage from current %u %u, igore\n", apReceivedTrans->m_TransId, apInfoSet->m_LastTrans.m_TransId );
		return;
	}

	// different stage
	if(EnumPrepareRejectResponse == apInfoSet->m_LastTrans.m_Type)
	{
		LOG("received action req but already reject from current %u %u, igore\n", apReceivedTrans->m_TransId, apInfoSet->m_LastTrans.m_TransId );
		return;
	}

	if(EnumPrepareOKResponse == apInfoSet->m_LastTrans.m_Type)
	{
		apInfoSet->m_LastTrans.m_TransId = apReceivedTrans->m_TransId;
		apInfoSet->m_LastTrans.m_NodeId = m_Info.m_Id;
		apInfoSet->m_LastTrans.m_Size = sizeof(Trans);

		leveldb::WriteBatch lBatch;

		if(EnumCommitRequest == apReceivedTrans->m_Type)
		{
			apInfoSet->m_LastTrans.m_Type = EnumCommitAck;
			m_pExecutor->Commit(apInfoSet->m_LastTrans.m_TransId, apInfoSet->m_TransData.c_str(), apInfoSet->m_TransData.length(), &lBatch);
		}
		else
		{
			apInfoSet->m_LastTrans.m_Type = EnumAbortAck;
			m_pExecutor->Abort(apInfoSet->m_LastTrans.m_TransId, apInfoSet->m_TransData.c_str(), apInfoSet->m_TransData.length(), &lBatch);
		}


		PC2DataStore::WriteLastTransState(&lBatch, apInfoSet);
		PC2DataStore::FinishAndClearTransactionInfo(&lBatch, apInfoSet);


		m_pStore->CommitBatch(&lBatch);
	}
	else // EnumCommitAck or EnumAbortAck
	{
		LOG("received action_req, already done trans_id %u\n", apInfoSet->m_LastTrans.m_TransId);
	}
}

void PC2Participant::SetExecutor( ParticipantExecutor* apExecutor )
{
	m_pExecutor = apExecutor;
}

void PC2Participant::SendDecisionRequest(uint32_t aTransID, std::set<uint32_t>* apParts)
{
	Trans lDecesionRequestTrans;
	lDecesionRequestTrans.m_NodeId = m_Info.m_Id;
	lDecesionRequestTrans.m_Size = sizeof(Trans);
	lDecesionRequestTrans.m_Type = EnumDecisionRequest;
	lDecesionRequestTrans.m_TransId = aTransID;
	NodeInfo* lpNode;

	lpNode = &m_CoordinatorInfo;
	SendTrans(lpNode, &lDecesionRequestTrans);

	if(NULL == apParts)
	{
		return;
	}

	for (auto it = apParts->begin(); it != apParts->end(); it++)
	{
		if(*it == m_Info.m_Id)
		{
			continue;
		}

		auto it2 = m_Parts.find(*it);
		if(it2 == m_Parts.end())
		{
			continue;
		}

		lpNode = &it2->second;
		SendTrans(lpNode, &lDecesionRequestTrans);
	}
	
}

}